package net.bwie.realtime.jtp.dws.log2.function2;

import net.bwie.realtime.jtp.dws.log2.bean2.PageViewBean2;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class PageViewReportWindowFunction2 implements WindowFunction<PageViewBean2,String,String, TimeWindow> {
    @Override
    public void apply(String key,
                      TimeWindow window,
                      Iterable<PageViewBean2> input,
                      Collector<String> out) throws Exception {

        //1.窗口函数
        String windowStart = DateTimeUtil.convertLongToString(window.getStart(),"yyyy-MM-dd HH:mm:ss");
        String windowEnd = DateTimeUtil.convertLongToString(window.getEnd(),"yyyy-MM-dd HH:mm:ss");

        //2.获取窗口中计算结果
        PageViewBean2 bean2 = input.iterator().next();
        Long pvCount = bean2.getPvCount();
        Long pvDuringTime = bean2.getPvDuringTime();
        Long uvCount = bean2.getUvCount();
        Long sessionCount = bean2.getSessionCount();

        //3.输出
        String output=windowStart+","+windowEnd+","+key
                +","+pvCount+","+pvDuringTime+","+uvCount+","+sessionCount
                +","+System.currentTimeMillis();

        out.collect(output);
    }
}
